/**
 * @author: yasen
 * @time: 2022/10/11 10:56
 * @email: yasen@yasenstudio.com
 */
package com.showdor.springboot;

import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.support.Acknowledgment;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;

public class ConsumerRecordDAOImpl implements ConsumerRecordDAO {
    private Connection connection;
    private PreparedStatement preparedStatement;

    public ConsumerRecordDAOImpl(Connection connection) {
        this.connection = connection;
    }

    private static Logger logger = LoggerFactory.getLogger(ConsumerRecordDAOImpl.class);

    @Override
    public <K> void create(K key, List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        String sql = "INSERT INTO `netcross`.motor_vehicle (MotorVehicleID, InfoKind,SourceID,TollgateID,DeviceID,StorageUrl1,StorageUrl2,StorageUrl3,StorageUrl4,StorageUrl5,MarkTime,LaneNo,PlateClass,PlateColor,PlateNo,PlateNoAttach,PlateDescribe,Speed,Direction,UsingPropertiesCode,VehicleClass,VehicleBrand,VehicleModel,VehicleColor,create_time,PassTime,NameOfPassedRoad,ThirdId) " +
                "VALUES (?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?);";
        logger.info("此线程消费" + records.size() + "条消息----线程名:" + Thread.currentThread().getName());
        try {
            preparedStatement = connection.prepareStatement(sql);
            for (ConsumerRecord<String, String> record : records) {
                //logger.info("topic名称:" + record.topic() + "\n" + "分区位置:" + record.partition() + "\n" + "key:" + record.key() + "\n" + "偏移量:" + record.offset() + "\n" + "消息内容:" + record.value());
                JSONObject valObj;
                try {
                    valObj = JSONObject.parseObject(record.value().toString());
                    if (valObj == null || !valObj.containsKey("data")) {
                        logger.error("数据为空，不合规" + record.value().toString());
                        ack.acknowledge();
                        continue;
                    }
                } catch (Exception e) {
                    logger.info("数据不合规：{}", e.getMessage());
                    logger.info("topic名称:" + record.topic() + "\n" + "分区位置:" + record.partition() + "\n" + "key:" + record.key() + "\n" + "偏移量:" + record.offset() + "\n" + "消息内容:" + record.value());
                    ack.acknowledge();
                    continue;
                }
                try {
                    JSONArray jsonArray = valObj.getJSONArray("data");
                    if (jsonArray == null) {
                        logger.info("获取data为null:", record.value());
                        continue;
                    }
                    JSONObject data = jsonArray.getJSONObject(0);
                    preparedStatement.setString(1, data.getString("MotorVehicleID"));
                    preparedStatement.setString(2, data.getString("InfoKind"));
                    preparedStatement.setString(3, data.getString("SourceID"));
                    preparedStatement.setString(4, data.getString("TollgateID"));
                    preparedStatement.setString(5, data.getString("DeviceID"));
                    preparedStatement.setString(6, data.getString("StorageUrl1"));
                    preparedStatement.setString(7, data.getString("StorageUrl2"));
                    preparedStatement.setString(8, data.getString("StorageUrl3"));
                    preparedStatement.setString(9, data.getString("StorageUrl4"));
                    preparedStatement.setString(10, data.getString("StorageUrl5"));
                    preparedStatement.setString(11, data.getString("MarkTime"));
                    preparedStatement.setInt(12, data.getIntValue("LaneNo", 0));
                    preparedStatement.setString(13, data.getString("PlateClass"));
                    preparedStatement.setString(14, data.getString("PlateColor"));
                    preparedStatement.setString(15, data.getString("PlateNo"));
                    preparedStatement.setString(16, data.getString("PlateNoAttach"));
                    preparedStatement.setString(17, data.getString("PlateDescribe"));
                    preparedStatement.setDouble(18, data.getDoubleValue("Speed"));
                    preparedStatement.setString(19, data.getString("Direction"));
                    preparedStatement.setString(20, data.getString("UsingPropertiesCode"));
                    preparedStatement.setString(21, data.getString("VehicleClass"));
                    preparedStatement.setString(22, data.getString("VehicleBrand"));
                    preparedStatement.setString(23, data.getString("VehicleModel"));
                    preparedStatement.setString(24, data.getString("VehicleColor"));
                    preparedStatement.setString(25, data.getString("create_time"));
                    preparedStatement.setString(26, data.getString("PassTime"));
                    preparedStatement.setString(27, data.getString("NameOfPassedRoad"));
                    preparedStatement.setString(28, data.getString("ThirdId"));
                    preparedStatement.addBatch();
                } catch (SQLException e) {
                    throw new RuntimeException(e);
                }
            }
            preparedStatement.executeBatch();
            ack.acknowledge();
        } catch (SQLException e) {
            logger.info("发生错误：{},{}", e.getMessage());
            e.printStackTrace();
        }finally {
            ack.acknowledge();
        }
    }

    @Override
    public <K, V> void create(K key, V value, Acknowledgment ack) {
        //logger.error("数据:" + value.toString());
        JSONObject valObj = null;
        try {
            valObj = JSONObject.parseObject(value.toString());
            if (valObj == null || !valObj.containsKey("data")) {
                logger.error("数据为空，不合规" + value.toString());
                ack.acknowledge();
                return;
            }
        } catch (Exception e) {
            logger.info("数据不合规：{}", e.getMessage());
            ack.acknowledge();
            return;
        }

        JSONObject data = valObj.getJSONArray("data").getJSONObject(0);
        try {
            String sql = "INSERT INTO `netcross`.motor_vehicle (MotorVehicleID, InfoKind,SourceID,TollgateID,DeviceID,StorageUrl1,StorageUrl2,StorageUrl3,StorageUrl4,StorageUrl5,MarkTime,LaneNo,PlateClass,PlateColor,PlateNo,PlateNoAttach,PlateDescribe,Speed,Direction,UsingPropertiesCode,VehicleClass,VehicleBrand,VehicleModel,VehicleColor,create_time,PassTime,NameOfPassedRoad,ThirdId) " +
                    "VALUES (?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?,?, ?);";
            preparedStatement = connection.prepareStatement(sql);
            preparedStatement.setString(1, data.getString("MotorVehicleID"));
            preparedStatement.setString(2, data.getString("InfoKind"));
            preparedStatement.setString(3, data.getString("SourceID"));
            preparedStatement.setString(4, data.getString("TollgateID"));
            preparedStatement.setString(5, data.getString("DeviceID"));
            preparedStatement.setString(6, data.getString("StorageUrl1"));
            preparedStatement.setString(7, data.getString("StorageUrl2"));
            preparedStatement.setString(8, data.getString("StorageUrl3"));
            preparedStatement.setString(9, data.getString("StorageUrl4"));
            preparedStatement.setString(10, data.getString("StorageUrl5"));
            preparedStatement.setString(11, data.getString("MarkTime"));
            preparedStatement.setInt(12, data.getIntValue("LaneNo", 0));
            preparedStatement.setString(13, data.getString("PlateClass"));
            preparedStatement.setString(14, data.getString("PlateColor"));
            preparedStatement.setString(15, data.getString("PlateNo"));
            preparedStatement.setString(16, data.getString("PlateNoAttach"));
            preparedStatement.setString(17, data.getString("PlateDescribe"));
            preparedStatement.setDouble(18, data.getDoubleValue("Speed"));
            preparedStatement.setString(19, data.getString("Direction"));
            preparedStatement.setString(20, data.getString("UsingPropertiesCode"));
            preparedStatement.setString(21, data.getString("VehicleClass"));
            preparedStatement.setString(22, data.getString("VehicleBrand"));
            preparedStatement.setString(23, data.getString("VehicleModel"));
            preparedStatement.setString(24, data.getString("VehicleColor"));
            preparedStatement.setString(25, data.getString("create_time"));
            preparedStatement.setString(26, data.getString("PassTime"));
            preparedStatement.setString(27, data.getString("NameOfPassedRoad"));
            preparedStatement.setString(28, data.getString("ThirdId"));
            preparedStatement.execute();
            ack.acknowledge();
        } catch (SQLException e) {
            //[INFO][2023-07-05 10:23:08][ConsumerRecordDAOImpl:68] - 发生错误：No value specified for parameter 14,{"MotorVehicleID":"1965924","InfoKind":"1","SourceID":"1","TollgateID":"1","DeviceID":"1","StorageUrl1":"http://192.168.139.185:801/offsiteViolation/batchUpload/undefined/浙A123457_01.jpg","StorageUrl2":"http://192.168.139.185:801/offsiteViolation/batchUpload/undefined/浙A123457_01.jpg","StorageUrl3":"1","StorageUrl4":"1","StorageUrl5":"1","MarkTime":"","LaneNo":"1","PlateClass":"1","PlateColor":"1","PlateNo":"1","PlateNoAttach":"1","PlateDescribe":"1","Speed":"1.0","Direction":"1","UsingPropertiesCode":"1","VehicleClass":"1","VehicleBrand":"1","VehicleModel":"1","VehicleColor":"1","create_time":"2023-07-04 10:09:55","NameOfPassedRoad":"1","ThirdId":"1"}
            logger.info("发生错误：{},{}", e.getMessage(), data.toString());
            e.printStackTrace();
        }
    }
}
